package de.preissler.createview;

import me.prettyprint.cassandra.model.BasicColumnFamilyDefinition;
import me.prettyprint.cassandra.service.ThriftCfDef;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
import me.prettyprint.hector.api.ddl.ColumnType;
import me.prettyprint.hector.api.ddl.ComparatorType;
import me.prettyprint.hector.api.factory.HFactory;

/**
 * Erstellt die materalsierten Sichten
 * @author Eric Preissler
 *
 */
public class Main 
{
	public static final String KEYSPACE_NAME = "Exchange";
	public static final String SCF_QUOTES_BY_CHANGE = "Quotes_by_Change";
	public static final String SCF_RVT_JOIN = "Join_Quotes_Dax";
	
	/**
	 * Start des Programms
	 * @param args Parameter werden ignoriert
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception
	{
		//Cluster initialisieren
		Cluster cluster = getCluster();
		
		MyUtil.trace("Materialized views will now be created!");
		
		//Die Sichten definieren
		createSchema(cluster);
		
		Keyspace keyspace = HFactory.createKeyspace(KEYSPACE_NAME, cluster);
		
		//RVT fuer QuotesByChange
		QuotesByChange quotesByChange = new QuotesByChange(keyspace);
		quotesByChange.createIndex();
		
		//RVT fuer Quotes und DAX als Join
		JoinQuotesDAX join = new JoinQuotesDAX(keyspace);
		join.createIndex();
		
		//LVT fuer vorberechneten DAX
		DAXPreCalc calcDAX = new DAXPreCalc(keyspace);
		calcDAX.createIndex();
		
		//Hector sauber beenden
		cluster.getConnectionManager().shutdown();
	}
	
	/*
	 * Aktualisiert das Schema der Cassandra Datenbank
	 */
	private static void createSchema(Cluster cluster)
	{
		BasicColumnFamilyDefinition definition = new BasicColumnFamilyDefinition();
		definition.setKeyspaceName(KEYSPACE_NAME);
		definition.setName(SCF_QUOTES_BY_CHANGE);
		definition.setComparatorType(ComparatorType.LONGTYPE);
		definition.setColumnType(ColumnType.SUPER);
		definition.setSubComparatorType(ComparatorType.UTF8TYPE);
		definition.setGcGraceSeconds(0);
        ColumnFamilyDefinition def = new ThriftCfDef(definition);
        //cluster.addColumnFamily(def);
        
		definition = new BasicColumnFamilyDefinition();
		definition.setKeyspaceName(KEYSPACE_NAME);
		definition.setName(Main.SCF_RVT_JOIN);
		definition.setComparatorType(ComparatorType.UTF8TYPE);
		definition.setColumnType(ColumnType.SUPER);
		definition.setSubComparatorType(ComparatorType.UTF8TYPE);
		definition.setGcGraceSeconds(0);
        def = new ThriftCfDef(definition);
        cluster.addColumnFamily(def);
        
        MyUtil.trace("Keyspace \"" + KEYSPACE_NAME + "\" has been updated!");
	}
	
	/*
	 * Erstellt das Cluster Objekt
	 */
	private static Cluster getCluster() throws Exception
	{
		String clusterName;
		String host;
		String port;
		
		System.out.println("In order to connect to cassandra we need some parameters...");
		
		System.out.print("clustername: ");
		clusterName = MyUtil.readLine();
		
		System.out.print("host: ");
		host = MyUtil.readLine();
		
		System.out.print("port: ");
		port = MyUtil.readLine();
		
		return HFactory.getOrCreateCluster(clusterName, host + ":" + port);
	}

}
